package com.cheng.im.service.mq;

import cn.hutool.core.util.ObjectUtil;
import com.cheng.im.common.bean.msg.ProtoMsgV1;
import com.cheng.im.common.constants.MQConstants;
import com.cheng.im.common.entity.SessionCache;
import com.cheng.im.common.protobuiler.MessageBuilder;
import com.cheng.im.service.util.UserSessionUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.catalina.manager.util.SessionUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

/**
 * mq生产者，系统消息推送到消息队列交由业务服务落盘
 * @author Cheng
 */
@Component
@Slf4j
public class RocketMQProduce {

    @Resource
    private RocketMQTemplate rocketMQTemplate;
    @Resource
    private UserSessionUtils userSessionUtils;

    /**
     * 发送消息到队列，注意：发送的消息得确保消息的session_id已经是接收方的session_id，不然消息发送过去也处理不了
     * @param top 发送的主题
     * @param pkg 发送的消息
     * @param brokerId 接收消息的节点
     * @return
     */
    public boolean sendMessage(String top, ProtoMsgV1.Message pkg, String brokerId) {
        // log.info("========消息队列:{}==========", top);
        String to = converString(brokerId);
        try {
            Message message = new Message(top+to, null, null, pkg.toByteArray());
            rocketMQTemplate.getProducer().send(message);
        } catch (MQClientException | MQBrokerException | RemotingException | InterruptedException e) {
            log.info("MQ发送出错");
            throw new RuntimeException(e);
        }
        return true;
    }

    public boolean sendMessage(String top, ProtoMsgV1.Message pkg) {
        log.info("========消息队列:{}==========", top);
        try {
            Message message = new Message(top, null, null, pkg.toByteArray());

            rocketMQTemplate.getProducer().send(message);
        } catch (MQClientException | MQBrokerException | RemotingException | InterruptedException e) {
            log.info("MQ发送出错");
            throw new RuntimeException(e);
        }
        return true;
    }

    public void sendToUser(String to, ProtoMsgV1.Message message){
        this.sendToUser(to, message, false);
    }

    public void sendToUser(String to, ProtoMsgV1.Message message, boolean isCreateNewPkg){
        ProtoMsgV1.Message sendMsg = message;
        SessionCache userSession = userSessionUtils.getUserSession(message.getAppId(), to, message.getSessionId());

        if (ObjectUtil.isNull(userSession)){
            return;
        }

        if (isCreateNewPkg) {
            sendMsg = MessageBuilder.updateSessionId(message, userSession.getSessionId());
        }

        sendMessage(MQConstants.SERVICE_2_IM, sendMsg, userSession.getImNode().getBrokerId());
    }

    /**
     * 发送除了from端以外的其他端
     * @param from 发送者id
     * @param message 消息
     */
    public void sendToUserExceptClient(String from, ProtoMsgV1.Message message){
        // log.info("同步消息到其他端");
        Map<String, SessionCache> userSession = userSessionUtils.getUserSession(message.getAppId(), from);

        userSession.values().stream().forEach(sessionCache -> {
            if (sessionCache.getSessionId().equals(message.getSessionId())){
                return;
            }
            // 将session_id切换为接收方的session_id
            ProtoMsgV1.Message otherClientMsg = MessageBuilder.updateSessionId(message, sessionCache.getSessionId());
            sendMessage(MQConstants.SERVICE_2_IM, otherClientMsg, sessionCache.getImNode().getBrokerId());
        });

    }

    private String converString(String source){
        return source.replace(".", "-").replace(":", "_");
    }

    public Map<String, SessionCache> dispatchMessage(String to, ProtoMsgV1.Message message) {
        // log.info("分发消息到其他用户端");
        Map<String, SessionCache> userSession = userSessionUtils.getUserSession(message.getAppId(), to);

        if (CollectionUtils.isEmpty(userSession)) {
            return null;
        }
        userSession.values().stream().forEach(sessionCache -> {
            // 将session_id切换为接收方的session_id
            ProtoMsgV1.Message otherClientMsg = MessageBuilder.updateSessionId(message, sessionCache.getSessionId());
            sendMessage(MQConstants.SERVICE_2_IM, otherClientMsg, sessionCache.getImNode().getBrokerId());
        });

        return userSession;
    }
}
